//! # 消息订阅者
//!
//! 定义和实现消息订阅者相关功能

use async_trait::async_trait;
use std::sync::Arc;
use tokio::sync::mpsc;

use crate::bus::{Subscriber, SubscriberStats};
use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;

/// 订阅者配置
#[derive(Debug, Clone)]
pub struct SubscriberConfig {
    /// 缓冲区大小
    pub buffer_size: usize,
    /// 是否自动确认消息
    pub auto_ack: bool,
    /// 最大未确认消息数
    pub max_unacked: usize,
    /// 消息处理超时时间（毫秒）
    pub message_timeout_ms: u64,
    /// 预取数量
    pub prefetch_count: usize,
}

impl Default for SubscriberConfig {
    fn default() -> Self {
        Self {
            buffer_size: 1000,
            auto_ack: true,
            max_unacked: 100,
            message_timeout_ms: 30000, // 30秒
            prefetch_count: 10,
        }
    }
}

/// 增强的内存订阅者实现
pub struct EnhancedMemorySubscriber {
    pub subscription_id: String,
    pub pattern: String,
    pub config: SubscriberConfig,
    pub receiver: Arc<tokio::sync::Mutex<mpsc::Receiver<Message>>>,
    pub stats: Arc<tokio::sync::Mutex<SubscriberStats>>,
    pub is_active: Arc<std::sync::atomic::AtomicBool>,
    pub unacked_messages: Arc<tokio::sync::Mutex<std::collections::HashMap<uuid::Uuid, Message>>>,
}

impl EnhancedMemorySubscriber {
    pub fn new(
        subscription_id: String,
        pattern: String,
        config: SubscriberConfig,
        receiver: mpsc::Receiver<Message>,
    ) -> Self {
        Self {
            subscription_id,
            pattern,
            config,
            receiver: Arc::new(tokio::sync::Mutex::new(receiver)),
            stats: Arc::new(tokio::sync::Mutex::new(SubscriberStats {
                subscribed_at: chrono::Utc::now(),
                ..Default::default()
            })),
            is_active: Arc::new(std::sync::atomic::AtomicBool::new(true)),
            unacked_messages: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
        }
    }

    /// 检查是否达到最大未确认消息数
    async fn check_max_unacked(&self) -> bool {
        let unacked = self.unacked_messages.lock().await;
        unacked.len() >= self.config.max_unacked
    }

    /// 清理过期的未确认消息
    async fn cleanup_expired_unacked(&self) {
        let timeout = chrono::Duration::milliseconds(self.config.message_timeout_ms as i64);
        let now = chrono::Utc::now();
        
        let mut unacked = self.unacked_messages.lock().await;
        unacked.retain(|_, message| {
            if let Some(received_at) = message.metadata.received_at {
                now.signed_duration_since(received_at) < timeout
            } else {
                true
            }
        });
    }
}

#[async_trait]
impl Subscriber for EnhancedMemorySubscriber {
    async fn try_receive(&self) -> MessageBusResult<Option<Message>> {
        if !self.is_active() {
            return Ok(None);
        }

        // 检查是否达到最大未确认消息数
        if !self.config.auto_ack && self.check_max_unacked().await {
            return Ok(None);
        }

        // 清理过期的未确认消息
        self.cleanup_expired_unacked().await;

        let mut receiver = self.receiver.lock().await;
        match receiver.try_recv() {
            Ok(mut message) => {
                message.metadata_mut().mark_received();
                
                // 如果不是自动确认，添加到未确认列表
                if !self.config.auto_ack {
                    let mut unacked = self.unacked_messages.lock().await;
                    unacked.insert(message.id(), message.clone());
                }
                
                let mut stats = self.stats.lock().await;
                stats.total_received += 1;
                stats.last_received_at = Some(chrono::Utc::now());
                stats.queue_length = receiver.len();
                
                Ok(Some(message))
            }
            Err(mpsc::error::TryRecvError::Empty) => Ok(None),
            Err(mpsc::error::TryRecvError::Disconnected) => {
                self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
                Ok(None)
            }
        }
    }

    async fn receive(&self) -> MessageBusResult<Option<Message>> {
        if !self.is_active() {
            return Ok(None);
        }

        // 检查是否达到最大未确认消息数
        if !self.config.auto_ack && self.check_max_unacked().await {
            // 等待一段时间后重试
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            return Ok(None);
        }

        // 清理过期的未确认消息
        self.cleanup_expired_unacked().await;

        let mut receiver = self.receiver.lock().await;
        match receiver.recv().await {
            Some(mut message) => {
                message.metadata_mut().mark_received();
                
                // 如果不是自动确认，添加到未确认列表
                if !self.config.auto_ack {
                    let mut unacked = self.unacked_messages.lock().await;
                    unacked.insert(message.id(), message.clone());
                }
                
                let mut stats = self.stats.lock().await;
                stats.total_received += 1;
                stats.last_received_at = Some(chrono::Utc::now());
                stats.queue_length = receiver.len();
                
                Ok(Some(message))
            }
            None => {
                self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
                Ok(None)
            }
        }
    }

    async fn receive_timeout(&self, timeout_ms: u64) -> MessageBusResult<Option<Message>> {
        if !self.is_active() {
            return Ok(None);
        }

        // 检查是否达到最大未确认消息数
        if !self.config.auto_ack && self.check_max_unacked().await {
            return Ok(None);
        }

        let timeout = tokio::time::Duration::from_millis(timeout_ms);
        let mut receiver = self.receiver.lock().await;
        
        match tokio::time::timeout(timeout, receiver.recv()).await {
            Ok(Some(mut message)) => {
                message.metadata_mut().mark_received();
                
                // 如果不是自动确认，添加到未确认列表
                if !self.config.auto_ack {
                    let mut unacked = self.unacked_messages.lock().await;
                    unacked.insert(message.id(), message.clone());
                }
                
                let mut stats = self.stats.lock().await;
                stats.total_received += 1;
                stats.last_received_at = Some(chrono::Utc::now());
                stats.queue_length = receiver.len();
                
                Ok(Some(message))
            }
            Ok(None) => {
                self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
                Ok(None)
            }
            Err(_) => Ok(None), // 超时
        }
    }

    async fn ack(&self, message: &Message) -> MessageBusResult<()> {
        if self.config.auto_ack {
            // 自动确认模式下，不需要手动确认
            return Ok(());
        }

        let mut unacked = self.unacked_messages.lock().await;
        if unacked.remove(&message.id()).is_some() {
            let mut stats = self.stats.lock().await;
            stats.total_acked += 1;
            Ok(())
        } else {
            Err(MessageBusError::receive_error("消息未找到或已确认"))
        }
    }

    async fn nack(&self, message: &Message, requeue: bool) -> MessageBusResult<()> {
        if self.config.auto_ack {
            // 自动确认模式下，不支持 nack
            return Err(MessageBusError::unsupported_operation("nack in auto-ack mode"));
        }

        let mut unacked = self.unacked_messages.lock().await;
        if unacked.remove(&message.id()).is_some() {
            let mut stats = self.stats.lock().await;
            stats.total_nacked += 1;
            
            if requeue {
                // TODO: 实现重新入队逻辑
                tracing::warn!("消息重新入队功能待实现: {}", message.id());
            }
            
            Ok(())
        } else {
            Err(MessageBusError::receive_error("消息未找到或已处理"))
        }
    }

    fn subscription_id(&self) -> &str {
        &self.subscription_id
    }

    fn pattern(&self) -> &str {
        &self.pattern
    }

    fn is_active(&self) -> bool {
        self.is_active.load(std::sync::atomic::Ordering::Relaxed)
    }

    async fn close(&self) -> MessageBusResult<()> {
        self.is_active.store(false, std::sync::atomic::Ordering::Relaxed);
        
        // 确认所有未确认的消息
        if !self.config.auto_ack {
            let mut unacked = self.unacked_messages.lock().await;
            let count = unacked.len();
            unacked.clear();
            
            if count > 0 {
                tracing::warn!("订阅者关闭时还有 {} 条未确认消息", count);
            }
        }
        
        Ok(())
    }

    async fn stats(&self) -> MessageBusResult<SubscriberStats> {
        let mut stats = self.stats.lock().await;
        
        // 更新当前队列长度
        let receiver = self.receiver.lock().await;
        stats.queue_length = receiver.len();
        drop(receiver);
        
        // 计算平均处理时间
        if stats.total_acked > 0 {
            let total_time = chrono::Utc::now()
                .signed_duration_since(stats.subscribed_at)
                .num_milliseconds() as f64;
            stats.average_processing_time_ms = total_time / stats.total_acked as f64;
        }
        
        Ok(stats.clone())
    }
}

/// 消息处理器接口
#[async_trait]
pub trait MessageHandler: Send + Sync {
    /// 处理消息
    ///
    /// # 参数
    /// - `message`: 要处理的消息
    ///
    /// # 返回
    /// - `Ok(())`: 处理成功
    /// - `Err(MessageBusError)`: 处理失败
    async fn handle(&self, message: Message) -> MessageBusResult<()>;

    /// 获取处理器名称
    fn name(&self) -> &str;
}

/// 自动处理订阅者 - 自动调用处理器处理消息
pub struct AutoProcessingSubscriber {
    subscriber: Arc<tokio::sync::Mutex<Box<dyn Subscriber>>>,
    handler: Arc<dyn MessageHandler>,
    is_running: Arc<std::sync::atomic::AtomicBool>,
    handle: Option<tokio::task::JoinHandle<()>>,
}

impl AutoProcessingSubscriber {
    /// 创建自动处理订阅者
    pub fn new(
        subscriber: Box<dyn Subscriber>,
        handler: Arc<dyn MessageHandler>,
    ) -> Self {
        let is_running = Arc::new(std::sync::atomic::AtomicBool::new(false));
        
        Self {
            subscriber: Arc::new(tokio::sync::Mutex::new(subscriber)),
            handler,
            is_running,
            handle: None,
        }
    }

    /// 启动自动处理
    pub fn start(&mut self) {
        self.is_running.store(true, std::sync::atomic::Ordering::Relaxed);
        
        let subscriber = Arc::clone(&self.subscriber);
        let handler = Arc::clone(&self.handler);
        let is_running = Arc::clone(&self.is_running);
        
        let handle = tokio::spawn(async move {
            while is_running.load(std::sync::atomic::Ordering::Relaxed) {
                let message = {
                    let subscriber_guard = subscriber.lock().await;
                    subscriber_guard.receive_timeout(100).await
                };
                
                match message {
                    Ok(Some(msg)) => {
                        if let Err(e) = handler.handle(msg).await {
                            tracing::error!("消息处理失败: {}", e);
                        }
                    }
                    Ok(None) => {
                        // 超时或没有消息，继续等待
                        continue;
                    }
                    Err(e) => {
                        tracing::error!("接收消息失败: {}", e);
                        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                    }
                }
            }
        });
        
        self.handle = Some(handle);
    }

    /// 停止自动处理
    pub async fn stop(&mut self) -> MessageBusResult<()> {
        self.is_running.store(false, std::sync::atomic::Ordering::Relaxed);
        
        if let Some(handle) = self.handle.take() {
            handle.await.map_err(|e| MessageBusError::internal_error(e.to_string()))?;
        }
        
        Ok(())
    }

    /// 检查是否正在运行
    pub fn is_running(&self) -> bool {
        self.is_running.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// 获取处理器
    pub fn handler(&self) -> &dyn MessageHandler {
        self.handler.as_ref()
    }
}

/// 简单消息处理器实现
pub struct SimpleMessageHandler {
    name: String,
    handler_fn: Arc<dyn Fn(Message) -> MessageBusResult<()> + Send + Sync>,
}

impl SimpleMessageHandler {
    /// 创建简单消息处理器
    pub fn new<F>(name: String, handler_fn: F) -> Self 
    where
        F: Fn(Message) -> MessageBusResult<()> + Send + Sync + 'static,
    {
        Self {
            name,
            handler_fn: Arc::new(handler_fn),
        }
    }
}

#[async_trait]
impl MessageHandler for SimpleMessageHandler {
    async fn handle(&self, message: Message) -> MessageBusResult<()> {
        (self.handler_fn)(message)
    }

    fn name(&self) -> &str {
        &self.name
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::message::Message;
    use serde_json::json;
    use tokio::sync::mpsc;

    #[tokio::test]
    async fn test_enhanced_memory_subscriber() {
        let (sender, receiver) = mpsc::channel(100);
        let config = SubscriberConfig {
            auto_ack: false,
            max_unacked: 5,
            ..Default::default()
        };
        
        let subscriber = EnhancedMemorySubscriber::new(
            "test-subscription".to_string(),
            "test.*".to_string(),
            config,
            receiver,
        );

        assert_eq!(subscriber.subscription_id(), "test-subscription");
        assert_eq!(subscriber.pattern(), "test.*");
        assert!(subscriber.is_active());

        // 发送消息
        let message = Message::new("test.topic", json!({"test": "data"}));
        sender.send(message.clone()).await.unwrap();

        // 接收消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        let received_message = received.unwrap();
        assert_eq!(received_message.payload(), message.payload());

        // 测试确认
        assert!(subscriber.ack(&received_message).await.is_ok());

        // 测试统计信息
        let stats = subscriber.stats().await.unwrap();
        assert_eq!(stats.total_received, 1);
        assert_eq!(stats.total_acked, 1);
    }

    #[tokio::test]
    async fn test_simple_message_handler() {
        let handler = SimpleMessageHandler::new(
            "test-handler".to_string(),
            |message| {
                println!("处理消息: {}", message.topic());
                Ok(())
            },
        );

        assert_eq!(handler.name(), "test-handler");

        let message = Message::new("test.topic", json!({"test": "data"}));
        assert!(handler.handle(message).await.is_ok());
    }

    #[tokio::test]
    async fn test_auto_processing_subscriber() {
        let (sender, receiver) = mpsc::channel(100);
        let config = SubscriberConfig::default();
        
        let subscriber = Box::new(EnhancedMemorySubscriber::new(
            "auto-test".to_string(),
            "test.*".to_string(),
            config,
            receiver,
        ));

        let handler = Arc::new(SimpleMessageHandler::new(
            "auto-handler".to_string(),
            |message| {
                println!("自动处理消息: {}", message.topic());
                Ok(())
            },
        ));

        let mut auto_subscriber = AutoProcessingSubscriber::new(subscriber, handler);
        
        // 启动自动处理
        auto_subscriber.start();
        assert!(auto_subscriber.is_running());

        // 发送消息
        let message = Message::new("test.topic", json!({"test": "auto"}));
        sender.send(message).await.unwrap();

        // 等待处理
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        // 停止自动处理
        assert!(auto_subscriber.stop().await.is_ok());
        assert!(!auto_subscriber.is_running());
    }
}